package urlcounter;

import java.io.IOException;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;


public class Stage2MapReduce {
	
	public static class TupleCreatorMapper extends
			Mapper<LongWritable, Text, Text, LongWritable> {

		private LongWritable one = new LongWritable(1);

		public void map(LongWritable qid, Text line,
				Context context) throws IOException, InterruptedException {
			SortedSet<String> urls = new TreeSet<String>();
			String lineSplit[] = line.toString().split("\t")[1].split(" ");
			for(int i = 0; i< lineSplit.length; i++){
				urls.add(lineSplit[i]);
			}
			Object[] asArray = urls.toArray();
			for (int i = 0; i < asArray.length - 1; i++) {
				for (int j = i + 1; j < asArray.length; j++) {
					context.write(new Text(asArray[i].toString() + " " + asArray[j].toString()), one);
				}
			}
		}
	}

	public static class TupleCounterReducer extends
			Reducer<Text, LongWritable, Text, LongWritable> {

		private LongWritable result = new LongWritable();

		public void reduce(Text URLtuple, Iterable<LongWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (LongWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(URLtuple, result);
		}
	}
}
